package com.wudl.realproces.utils

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ColumnFamilyDescriptor, _}
import org.apache.hadoop.hbase.util.Bytes

/**
 * Hbase 的工具类
 */
object HBaseUtil {

  //  创建Hbase 的配置类
  val conf: Configuration = HBaseConfiguration.create()
  // Hbase 的连接
  private val conn: Connection = ConnectionFactory.createConnection(conf)
  // Hbase 的操作api
  private val admin: Admin = conn.getAdmin

  /**
   * 返回一个表， 如果比较不存在就 创建一个表
   *
   * @param tableNameStr
   * @param columnFamilyName
   * @return
   */

  def getTable(tableNameStr: String, columnFamilyName: String): Table = {

    // 获取tableName
    val tableName: TableName = TableName.valueOf(tableNameStr)
    // 如果表不能存在， 就创建表
    if (!admin.tableExists(tableName)) {
      val descBuilder: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName)
      val familyDescriptor: ColumnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyName.getBytes()).build()
      // 给表添加列族
      descBuilder.setColumnFamily(familyDescriptor)
      // 在创建表
      admin.createTable(descBuilder.build())
    }
    conn.getTable(tableName)
  }

  /**
   * 存储单列的数据
   *
   * @param tableNmaeStr     表名称
   * @param rowkey           hbase 的rowkey
   * @param columnFamilyName hbase 的列簇
   * @param columnName       列的名称
   * @param columnvalue      列的值
   */

  def putData(tableNmaeStr: String, rowkey: String, columnFamilyName: String, columnName: String, columnvalue: String): Unit = {
    // 获取表明
    val table: Table = getTable(tableNmaeStr, columnFamilyName)
    try {
      val put: Put = new Put(rowkey.getBytes)
      put.addColumn(columnFamilyName.getBytes, columnName.getBytes, columnvalue.getBytes)
      table.put(put);
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    } finally {
      table.close()
    }

  }

  /**
   * 通过列名获取列值
   *
   * @param tableNameStr
   * @param rowkey
   * @param columnFamilyName
   * @param columnName
   * @return
   */
  def getData(tableNameStr: String, rowkey: String, columnFamilyName: String, columnName: String): String = {
    // 1.获取table 对象
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      //2. 构建get 对象
      val get = new Get(rowkey.getBytes)
      // 3.进行查询
      val result: Result = table.get(get)
      //4.判断查询结果是否给空，并且包含我们查询的列
      if (result != null && result.containsColumn(columnFamilyName.getBytes, columnName.getBytes)) {
        val bytes: Array[Byte] = result.getValue(columnFamilyName.getBytes(), columnName.getBytes)
        Bytes.toString(bytes)
      } else {
        ""
      }
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        ""
      }
    } finally {
      // 5.关闭table
      table.close()
    }
  }


  /**
   * 存储多列数据
   *
   * @param tableNameStr
   * @param rowkey
   * @param columnFamilyName
   * @param map
   */
  def putMapData(tableNameStr: String, rowkey: String, columnFamilyName: String, map: Map[String, Any]): Unit = {
    //1.获取table
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      // 2.创建put
      val put = new Put(rowkey.getBytes)
      // 3. 在put 中添加多个列名和值
      for ((colName, colValue) <- map) {
        put.add(columnFamilyName.getBytes, colName.getBytes, colValue.toString.getBytes)
      }
      // 4. 保存put
      table.put(put)
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    } finally {
      // 5. 关闭table
      table.close()
    }
  }

  /**
   * 获取多列数据的值
   *
   * @param tableNameStr
   * @param rowkey
   * @param columnFamilyName
   * @param columnNameList
   * @return
   */
  def getMapData(tableNameStr: String, rowkey: String, columnFamilyName: String, columnNameList: List[String]): Map[String, String] = {

    //1. 获取table
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      //2.构建get
      val get = new Get(rowkey.getBytes)
      // 3.执行查询
      val result: Result = table.get(get)
      // 4.遍历列名集合， 取出值，构建map 返回
      columnNameList.map {
        col =>
          val bytes: Array[Byte] = result.getValue(columnFamilyName.getBytes, col.getBytes)
          if (bytes != null && bytes.size > 0) {
            col -> Byte.toString(bytes)
          } else {
            "" -> ""
          }
      }.filter(_._1 != "").toMap
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        Map[String, String]()
      }
    } finally {
      table.close()
    }
  }

  /**
   * 根据rowkey 来删除数据
   *
   * @param tablNameStr
   * @param rowkey
   * @param columnFamilyNameStr
   */
  def deleteData(tablNameStr: String, rowkey: String, columnFamilyNameStr: String): Unit = {

    // 1.获取table
    val table = getTable(tablNameStr, columnFamilyNameStr)
    try {
      // 2.构建delete对象
      val delete = new Delete(rowkey.getBytes)
      // 3.执行删除
      table.delete(delete)

    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    } finally {
      //4.关闭table
      table.close()
    }
  }
}
